package com.example.mq.mqserver.datacenter;

import com.example.mq.common.BinaryTool;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * 针对硬盘上的消息进行管理
 */
public class MessageFileManger {

    //定义一个内部类 表示该队列的统计信息
    static public class Stat{
        public int totalCount;  //总消息数
        public int validCount;  //有效信息数
    }

    //初始化方法
    public void init(){
        //暂时不需要做啥额外的操作 只是为了代码统一 如果后面拓展 可以在这拓展
    }

    //约定消息文件所在的目录和文件名
    //这个方法 获取指定队列 所在的文件所在的路径
    private String getQueueDir(String queueName){
        return "./data/"+queueName;
    }
    //用来获取该队列的消息数据文件路径
    //二进制文件 一般是使用.bin/.dat 来存储 这里我们使用txt不太合理 但是也可以
    private String getQueueDataPath(String queueName){
        return getQueueDir(queueName)+"/queue_data.txt";
    }

    //获取该队列的消息统计文件
    private String getQueueStatPath(String queueName){
        return getQueueDir(queueName)+"/queue_stat.txt";
    }

    //读取指定的统计文件
    private Stat readStat(String queueName){
        //由于当前的消息统计文件是文本文件 可以直接使用Scanner 来读取文件
        Stat stat=new Stat();
        try (InputStream inputStream=new FileInputStream(getQueueStatPath(queueName))){
            Scanner scanner=new Scanner(inputStream);
            stat.totalCount=scanner.nextInt();  //读到的int类型 为消息总数
            stat.validCount=scanner.nextInt();  //读取到的为有效消息的总数
            return stat;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }
    //写入统计文件
    private void writeStat(String queueName,Stat stat){
        //使用PrintWrite 来写文件
        //OutputStream 默认情况下 打开文件的时候 会默认将该文件清空 相当于新的数据覆盖了旧的数据
        try(OutputStream outputStream=new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter=new PrintWriter(outputStream);
            printWriter.write(stat.totalCount+"\t"+stat.validCount);  //写入数据
            printWriter.flush();  //刷新缓冲区
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //创建队列对应的文件和目录
    public void createQueueFiles(String queueName) throws IOException {
        //1.先创建队列对应的目录
        File baseDir=new File(getQueueDir(queueName));
        if(!baseDir.exists()){
            //不存在该目录就创建
            boolean ok=baseDir.mkdirs();
            if(!ok){
                //如果创建失败
                throw new IOException("创建目录失败 baseDir="+baseDir.getAbsolutePath());
            }
        }
        //2.创建队列数据文件
        File queueDataFile=new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()){
            //如果吧不存在该文件就创建文件
            boolean ok=queueDataFile.createNewFile();
            if(!ok){
                //创建文件失败 抛出异常
                throw new IOException("创建文件失败失败 queueDataFile="+queueDataFile.getAbsolutePath());
            }
        }
        //3.创建消息统计文件
        File queueStatFile=new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            //如果文件不存在
            boolean ok=queueStatFile.createNewFile();
            if(!ok){
                //创建文件失败 抛出异常
                throw new IOException("创建文件失败失败 queueStatFile="+queueStatFile.getAbsolutePath());
            }
        }
        //4.给消息统计文件 设置默认值为0\t0
        Stat stat=new Stat();
        stat.validCount=0;
        stat.totalCount=0;
        writeStat(queueName,stat);
    }
    //删除目录和文件
    //队列也是可以被删除的 删除队列后 对应的消息文件也对应删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除文件 在删除目录
        File queueDateFile=new File(getQueueDataPath(queueName));
        boolean ok =queueDateFile.delete();
        File queueStatFile=new File(getQueueStatPath(queueName));
        boolean ok2=queueStatFile.delete();
        File baseDir=new File(getQueueDir(queueName));
        boolean ok3=baseDir.delete();
        if(!ok||!ok2||!ok3){
            //如果删除失败
            throw new IOException("删除文件和文件失败! baseDir="+baseDir.getAbsolutePath());
        }
    }
    //检查队列的文件/目录是否存在
    public boolean checkFilesExits(String queueName){
        //队列的数据文件和统计文件是否都存在  如果有一方不存在 说明数据损坏
        File queueDateFile=new File(getQueueDataPath(queueName));
        if(!queueDateFile.exists()){
            return false;
        }
        File queueStatFile=new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()){
            return false;
        }
        return true;
    }

    //这个方法用于将消息 写入指定的队列
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.校验当前队列 对应的文件是否存在
        if (!checkFilesExits(queue.getName())) {
            //如果不存在 抛出自定义异常
            throw new MqException("[MessageFileManager] 队列对应的文件不存在 queueName=" + queue.getName());
        }
        //2.将Message对象序列化 转成二进制的字节数组
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue){
            //3.先获取到消息对象的数据长度 以此来确定Message的 offsetBeg 和offsetEnd
            //把新的Message数据 写到数据文件的末尾 此时Message 对象的 offsetBeg 就是当前文件长度 +4
            // (这里的4字节是我们之前规定的(也就是消息长度 占四个字节))
            //offsetEnd 就是当前Message自身长度+4+当前文件长度
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);//通过这个方法就可获取长度 单位为字节
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息到数据文件  注意数据写入是追加写 不是覆盖写 设置参数为true
            try (OutputStream outputStream = new FileOutputStream(queueDataFile, true)) {
                //写入当前消息的长度 占据4个字节 使用标准库提供的dataOutputStream 而我们的outputStream一次直只能写入一个字节
                try (DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //写入当前消息的长度 占据4个字节~
                    dataOutputStream.writeInt(messageBinary.length);
                    //接下来 写入消息体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.已经写入新的消息 更新消息统计文件
            Stat stat=readStat(queue.getName());
            stat.totalCount+=1;  //消息总数+1
            stat.validCount+=1;     //有效消息+1
            writeStat(queue.getName(),stat);
        }
    }

    //删除队列中的指定消息
    //此处的删除只是逻辑删除 存在文件中的 isValid 属性设置为无效
    //1.先把文件中的这一段数据,读出来 还原成Message对象
    //2.将对象的isValid对象设置为 0
    //3.将对象重写转换成二进制数据 写回文件
    //此处的参数中的Message对象 必须包含有效的offsetBeg和offsetEnd属性
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException, MqException {
        //之前用的都是FileOutputStream和 FileInputStream 都是从文件头开始写的
        //此处我们需要在指定的位置读取信息 随机访问
        //此处用到的类是  RandomAccessFile
        synchronized (queue){  //保证线程安全
            try(RandomAccessFile randomAccessFile=new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")){
                //1.先从文件中读取对应的Message数据
                byte[] bufferSrc=new byte[(int) (message.getOffsetEnd()- message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());
                //读取数据  从OffsetBeg 位置读取 一直读取到bufferSrc数组满为止
                randomAccessFile.read(bufferSrc);
                //将读出的数据 还原成Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                //将对象的isValid对象设置为 0
                diskMessage.setIsValid((byte) 0x0);
                //序列化对象
                byte[] bufferDest=BinaryTool.toBytes(diskMessage);
                //调整光标位置 防止因为读写 导致光标位置移动
                randomAccessFile.seek(message.getOffsetBeg());
                //将文件重新写回文件
                randomAccessFile.write(bufferDest);
            }
            //更新统计文件
            Stat stat=readStat(queue.getName());
            //这里无需将总数-1 因为我们是逻辑删除 只需要标记有多少个有效文件即可
            if(stat.validCount>0){
                stat.validCount-=1;     //有效消息-1
            }
            writeStat(queue.getName(),stat);
        }
    }

    /**
     * 从文件中读取队列中的全部消息 到内存中
     * 这个方法 准备在程序启动的时候 进行调用
     */
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages=new LinkedList<>();
        try(InputStream inputStream=new FileInputStream(getQueueDataPath(queueName))){
            //用这个变量来记录当前文件光标位置
            long currentOffSet=0;
            try(DataInputStream dataInputStream=new DataInputStream(inputStream)){
                //一个文件包含多个消息 需要循环处理
                while (true){
                    //1.读取4个字节的消息长度
                    //这里的 readInt 方法 可能会读取到文件的末尾
                    //readInt 方法 读到末尾 会抛出 EOFException 异常  这一点和之前的流对象不太一样
                    int messageSize=dataInputStream.readInt();
                    //2.按照长度 读取消息本体
                    byte[]  buffer=new byte[messageSize];
                    int actualSize=dataInputStream.read(buffer);
                    if(actualSize!=messageSize){
                        //预期消息长度 和实际读取消息长度 不相同  格式错乱了
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName="+queueName);
                    }
                    //3.把读到的数据 反序列化成Message对象
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    //4.判断一下消息对象是不是无效对象
                    if(message.getIsValid()!=0x1){
                        //无效数据 更新光标位置 跳过
                        currentOffSet+=(messageSize+4);
                        continue;
                    }
                    //5.添加消息到链表中  添加之前需要填写offsetBeg和offsetEnd
                    //在计算之前 需要知道当前文件光标的位置 但是当前dataInputStream无法获取光标位置
                    //因此需要我们手动计算
                    message.setOffsetBeg(currentOffSet+4);
                    message.setOffsetEnd(currentOffSet+4+messageSize);
                    //更新光标位置
                    currentOffSet+=(messageSize+4);
                    messages.add(message);
                }
            }catch (EOFException e){
                 //这个catch 并非是 真的发生异常了 而是"正常的业务逻辑" 文件读到末尾 会被 readInt 抛出异常
                //我们也不需要做特殊处理
                System.out.println("[MessageFileManager] 恢复 Message 数据完成");
             }
        }
        return messages;
    }

    /**
     * 检查当前是否要针对该队列的消息数据文件进行GC
     * @param queueName  队列名称
     * @return
     */
    public boolean checkGC(String queueName){
        //判断是否要GC:当总的消息数超过2000 并且有效数据低于总消息数目的50%,就触发一次GC
        //这两个信息都存在与消息统计文件中
        Stat stat=readStat(queueName);
        if(stat.totalCount > 2000&& (double)stat.validCount / (double) stat.totalCount < 0.5){
            return true;
        }
        return false;
    }

    //GC回收中的 重新创建新的文件名
    public String getQueueDataNewPath(String queueName){
        return getQueueDir(queueName)+"/queue_data_new.txt";
    }

    //通过这个方法 真正执行GC
    //使用复制算法
    //创建一个新的文件 将旧文件中的有效数据都复制到新文件  将旧文件的数据全部删除
    //删除旧的文件后 将新文件命名为 queue_data.txt
    //最后更新消息统计文件
    public  void  gc(MSGQueue queue) throws IOException, MqException, ClassNotFoundException {
        //进行 GC 的时候 其他线程不能对消息进行修改
        synchronized (queue){
            //由于 gc 可能比较耗时 此处我们统计一下执行消耗的时间
            long gcBeg=System.currentTimeMillis();
            //1.创建一个新文件
            File queueDataNewFile=new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()){
                //正常情况下  这个文件是不存在的 如果存在 计算意外 说明上次gc了一半
                throw new MqException("[MessageFileManager] gc时发现该队列的 queue_data_new已经存在 queueName="+queue.getName());
            }
            //如果不存在就创建文件
            boolean ok=queueDataNewFile.createNewFile();
            if(!ok){
                //创建失败
                throw new MqException("[MessageFileManager] 创建文件时失败 queueName="+queue.getName());
            }
            //将旧文件中的有效数据都复制到新文件
            //从硬盘中读取所有消息 并写入到新文件中
            //这里没有判断有效性 因为我们读取这个消息的时候 loadAllMessageFromQueue 进行了有效性的判断
            //所以保证我们读取到的数据都是有效数据 从而写入新文件的都是有效数据
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());
            try(OutputStream outputStream=new FileOutputStream(queueDataNewFile)){
                try(DataOutputStream dataOutputStream=new DataOutputStream(outputStream)){
                    for (int i = 0; i <messages.size(); i++) {
                        //将每个消息转换成二进制数据
                        byte[] buffer = BinaryTool.toBytes(messages.get(i));
                        //写入新文件  先写入四个字节的消息长度  再写入正文
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //删除旧的数据文件 并且把新的文件进行重命名
            File queueDataOldFile=new File(getQueueDataPath(queue.getName()));
            ok=queueDataOldFile.delete();
            if(!ok){
                //如果删除失败 抛出异常
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败 queueDataOldFile="+queueDataOldFile.getAbsolutePath());
            }
            //给新文件重命名
            ok=queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok){
                throw new MqException("[MessageFileManager] 文件重命名失败 queueDataOldFile="+queueDataOldFile+"queueDataNewFile"+queueDataNewFile);
            }
            //更新统计消息
            Stat stat=readStat(queue.getName());
            stat.totalCount=messages.size();
            stat.validCount=messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd=System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕 ! queueName="
                    +queue.getName()+", time="+(gcEnd-gcBeg)+"ms");
        }
    }

}

